from flask import Flask, send_from_directory
from flask_cors import CORS, cross_origin
from flask_socketio import SocketIO, emit
from kafka import KafkaProducer, KafkaConsumer, TopicPartition
import uuid
import docker
import os
import json

client = docker.from_env()
app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="*")
cors = CORS(app)
app.config['CORS_HEADERS'] = 'Content-Type'

BOOTSTRAP_SERVERS = 'kafka:9092'
TOPIC_NAME = 'images-download'
_mail_enabled = os.environ.get("MAIL_ENABLED", default="true")
_docker_registry = os.environ.get('DOCKER_REGISTRY',default="10.0.1.1:88")
_registry_username = os.environ.get('REGISTRY_USERNAME',default="admin")
_registry_password = os.environ.get('REGISTRY_PASSWORD',default="Harbor12345")


@app.route('/')
@cross_origin()
def home():
    return send_from_directory('/app', "index.html")

""" Kafka endpoints """


@socketio.on('connect', namespace='/kafka')
def test_connect():
    emit('logs', {'data': 'Connection established'})


@socketio.on('kafkaconsumer', namespace="/kafka")
def kafkaconsumer(message):

    consumer = KafkaConsumer(group_id='consumer-1',
                             bootstrap_servers=BOOTSTRAP_SERVERS)
    tp = TopicPartition(TOPIC_NAME, 0)
    # register to the topic
    consumer.assign([tp])

    # obtain the last offset value
    consumer.seek_to_end(tp)
    lastOffset = consumer.position(tp)
    consumer.seek_to_beginning(tp)
    emit('kafkaconsumer1', {'data': ''})
    for message in consumer:
        emit('kafkaconsumer', {'data': message.value.decode('utf-8')})
        if message.offset == lastOffset - 1:
            break
    consumer.close()


@socketio.on('kafkaproducer', namespace="/kafka")
def kafkaproducer(message):
    print(TOPIC_NAME, BOOTSTRAP_SERVERS, _docker_registry, _registry_username, _registry_password)
    splitLine(message)
    producer = KafkaProducer(bootstrap_servers=BOOTSTRAP_SERVERS)
    producer.send(TOPIC_NAME, value=bytes(str(message), encoding='utf-8'), key=bytes(str(uuid.uuid4()), encoding='utf-8'))
    emit('logs', {'data': 'Added ' + message + ' to topic'})
    emit('kafkaproducer', {'data': message})
    producer.close()
    kafkaconsumer(message)

def splitLine(images):
    imagessplit = images.split("\n")
    for image in imagessplit:
        if image.find(":") < 0:
            image += ':latest'
        original = image.split(":")[0]
        name = original.split("/")[-1]
        tag = image.split(":")[1]
        pulledImage = pullImage(original,tag)
        newimage = tagImage(pulledImage, tag, name)
        pushImage(newimage,tag)


def pullImage(image,tag):

    print(' start pull ' +image + ' : ' +tag)
    image = client.images.pull(image, tag)
    print(image)
    return image


def tagImage(image,tag,name):

    newimagename = _docker_registry+'/library/'+name
    image.tag(_docker_registry+'/library/'+name, tag)
    print(image.tags)
    return newimagename


def pushImage(image,tag):
    # loginReg()  # 仓库不支持api ssl登录,如果需要登录要使用ssl转发方式
    pushResult = client.api.push(image, tag)
    print(pushResult)
    return pushResult


def loginReg():
    return client.login(username=_registry_username, password=_registry_password, registry=_docker_registry)


if __name__ == '__main__':
    socketio.run(app, host='0.0.0.0', port=80)
